package com.k2data.platform.backup.consumer;

import org.apache.commons.compress.compressors.bzip2.BZip2CompressorInputStream;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.net.URI;
import java.util.zip.GZIPInputStream;

/**
 * This is demo for decompress files in hdfs.
 */
public class DecompressDemo {
    public static void main(String[] args) {
        String hdfsUrl = "hdfs://10.1.10.21:9000";
        String filePath = "/k2data/dataplatform/processing/abnormaldata/gw/exception4/group1/type2/0-1490005745854.bzip2.complete";
        Configuration conf = new Configuration();

        // workaround bug in maven assembly: http://stackoverflow.com/questions/17265002/hadoop-no-filesystem-for-scheme-file
        conf.set("fs.hdfs.impl",
                org.apache.hadoop.hdfs.DistributedFileSystem.class.getName()
        );
        conf.set("fs.file.impl",
                org.apache.hadoop.fs.LocalFileSystem.class.getName()
        );

        FileSystem fs = null;
        String codec = "bzip2";
        InputStream in = null; 
//        FSDataInputStream in = null;  //no compress
//        GZIPInputStream in = null;
//        BZip2CompressorInputStream in = null;
        BufferedReader br = null;
        byte[] buff = new byte[256];
        try {
            fs = FileSystem.get(URI.create(hdfsUrl), conf);
            if (codec.equals("nocomp")) {
              in = fs.open(new Path(filePath));
            } else if (codec.equals("bzip2")) {
              in = new BZip2CompressorInputStream(fs.open(new Path(filePath)));
            } else if (codec.equals("gzip")) {
              in = new GZIPInputStream(fs.open(new Path(filePath)));
            }
            br = new BufferedReader(new InputStreamReader(in));
            String data = null;
            while ((data = br.readLine()) != null) {
                System.out.println(data);
            }
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            if (br != null) {
                try {
                    br.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
            if (in != null) {
                try {
                    in.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
            if (fs != null) {
                try {
                    fs.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
    }
}
